package org.apache.rocketmq.broker.processor;

import io.netty.channel.ChannelHandlerContext;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.attribute.AttributeParser;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.remoting.CommandCustomHeader;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RemotingSysResponseCode;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.header.CreateTopicRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.DeleteTopicRequestHeader;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;

public class AdminBrokerProcessor implements NettyRequestProcessor {
    private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.BROKER_LOGGER_NAME);

    protected final BrokerController brokerController;

    //这个配置黑名单名称在第六版本代码还用不到，大家可以暂时忽略
    protected Set<String> configBlackList = new HashSet<>();

    //构造方法
    public AdminBrokerProcessor(final BrokerController brokerController) {
        this.brokerController = brokerController;
        //初始化配置黑名单
        initConfigBlackList();
    }



    //初始化配置黑名单的方法，这里我可以给大家简单解释一下，就是在这个黑名单中的配置是不能被修改的，这个是为了保证系统的稳定性，不让用户随意修改的
    //在源码中，当前类中有一个updateBrokerConfig()方法，如果用户使用控制台模块向Broker发送了修改配置的请求，也就是发送了RequestCode.UPDATE_BROKER_CONFIG类型的请求
    //那么就会调用这个updateBrokerConfig()方法处理该请求，在updateBrokerConfig()方法中会根据黑名单来判断是否可以修改对应的配置信息
    //比如brokerConfigPath、rocketmqHome这些信息都是不允许修改的
    private void initConfigBlackList() {
        //Broker配置文件的路径
        configBlackList.add("brokerConfigPath");
        //RocketMQ的Home路径
        configBlackList.add("rocketmqHome");
        //黑名单配置信息也不允许修改
        configBlackList.add("configBlackList");
        String[] configArray = brokerController.getBrokerConfig().getConfigBlackList().split(";");
        configBlackList.addAll(Arrays.asList(configArray));
    }
    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws Exception {
        switch (request.getCode()){
            case RequestCode.UPDATE_AND_CREATE_TOPIC:
                return updateAndCreateTopic(ctx,request);
            case RequestCode.DELETE_TOPIC_IN_BROKER:
                return deleteTopic(ctx,request);
            default:
                return getUnknownCmdResponse(ctx, request);
        }
    }

    private RemotingCommand deleteTopic(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        DeleteTopicRequestHeader requestHeader = (DeleteTopicRequestHeader) request.decodeCommandCustomHeader(DeleteTopicRequestHeader.class);
        LOGGER.info("AdminBrokerProcessor#deleteTopic: broker receive request to delete topic={}, caller={}",
                requestHeader.getTopic(), RemotingHelper.parseChannelRemoteAddr(ctx.channel()));

        //得到要删除的主题名称
        String topic = requestHeader.getTopic();
        //对主题名称执行判空操作
        if (UtilAll.isBlank(topic)) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("The specified topic is blank.");
            return response;
        }
        //判断要删除的主题信息是否为系统主题，如果是系统主题则直接返回错误响应
        if (brokerController.getBrokerConfig().isValidateSystemTopicWhenUpdateTopic()) {
            if (TopicValidator.isSystemTopic(topic)) {
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("The topic[" + topic + "] is conflict with system topic.");
                return response;
            }
        }
        //以下内容在第七版本代码才会用到，所以在第六版本代码中先注释掉
//        final Set<String> groups = this.brokerController.getConsumerOffsetManager().whichGroupByTopic(topic);
//        try {
//
//            for (String group : groups) {
//                final String popRetryTopicV2 = KeyBuilder.buildPopRetryTopic(topic, group, true);
//                if (brokerController.getTopicConfigManager().selectTopicConfig(popRetryTopicV2) != null) {
//                    deleteTopicInBroker(popRetryTopicV2);
//                }
//                final String popRetryTopicV1 = KeyBuilder.buildPopRetryTopicV1(topic, group);
//                if (brokerController.getTopicConfigManager().selectTopicConfig(popRetryTopicV1) != null) {
//                    deleteTopicInBroker(popRetryTopicV1);
//                }
//            }
//            deleteTopicInBroker(topic);
//        } catch (Throwable t) {
//            return buildErrorResponse(ResponseCode.SYSTEM_ERROR, t.getMessage());
//        }

        //设置成功响应
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

    private RemotingCommand updateAndCreateTopic(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        final CreateTopicRequestHeader requestHeader =
                (CreateTopicRequestHeader)request.decodeCommandCustomHeader(CreateTopicRequestHeader.class);
        String topic = requestHeader.getTopic();
        TopicValidator.ValidateTopicResult validateTopicResult = TopicValidator.validateTopic(topic);
        if (!validateTopicResult.isValid()) {
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark(validateTopicResult.getRemark());
            return response;
        }

        if (this.brokerController.getBrokerConfig().isValidateSystemTopicWhenUpdateTopic()){
            if (TopicValidator.isSystemTopic(topic)) {
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("The topic[" + topic + "] is a system topic.");
                return response;
            }
        }
        TopicConfig topicConfig = new TopicConfig( topic);
        topicConfig.setReadQueueNums(requestHeader.getReadQueueNums());
        topicConfig.setWriteQueueNums(requestHeader.getWriteQueueNums());
        topicConfig.setPerm(requestHeader.getPerm());
        topicConfig.setTopicFilterType(requestHeader.getTopicFilterTypeEnum());
        topicConfig.setTopicSysFlag(requestHeader.getTopicSysFlag());
        topicConfig.setOrder(requestHeader.getOrder());
        String attributes = requestHeader.getAttributes();
        topicConfig.setAttributes(AttributeParser.parseToMap( attributes));


        //得到该主题支持的消息类型，然后判断该类型是否和混合消息类型一致，如果一致则判断broker是否支持混合消息类型的主题，如果不支持则返回错误响应
        if (topicConfig.getTopicMessageType() == TopicMessageType.MIXED && !brokerController.getBrokerConfig().isEnableMixedMessageType()) {
            //源码中默认是不支持的，所以直接返回错误响应即可
            response.setCode(ResponseCode.SYSTEM_ERROR);
            //设置错误信息，Broker不支持混合消息类型
            response.setRemark("MIXED message type is not supported.");
            return response;
        }
        //如果一样则意味着当前Broker已经存储了对应的主题，那么直接返回成功响应即可
        if (topicConfig.equals(this.brokerController.getTopicConfigManager().getTopicConfigTable().get(topic))) {
            LOGGER.info("Broker receive request to update or create topic={}, but topicConfig has  no changes , so idempotent, caller address={}",
                    requestHeader.getTopic(), RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
            //设置成功响应
            response.setCode(ResponseCode.SUCCESS);
            return response;
        }
        try {
//            //下面就是根据主题的配置信息更新对应主题的操作
            this.brokerController.getTopicConfigManager().updateTopicConfig(topicConfig);
//            //判断是否使用单主题注册方式注册主题
            if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) {
                //单主题注册方式注册主题
                this.brokerController.registerSingleTopicAll(topicConfig);
            } else {
                //增量注册主题
                this.brokerController.registerIncrementBrokerData(topicConfig, this.brokerController.getTopicConfigManager().getDataVersion());
            }
            //设置成功响应
            response.setCode(ResponseCode.SUCCESS);
        } catch (Exception e) {
            LOGGER.error("Update / create topic failed for [{}]", request, e);
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark(e.getMessage());
        }
        return response;
    }

    //该方法在第七版本代码中才会真正实现
    private void deleteTopicInBroker(String topic) {

    }


    private RemotingCommand getUnknownCmdResponse(ChannelHandlerContext ctx, RemotingCommand request) {
        //构建不支持该请求类型的信息
        String error = " request type " + request.getCode() + " not supported";
        //创建响应对象
        final RemotingCommand response = RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
        return response;
    }

    @Override
    public boolean rejectRequest() {
        return false;
    }
}